package com.yujiahao.bigdata.teacher

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark03_Req_PageFlowAnalysis {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("HotCategoryTop10SessionAnalysis").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // TODO 需求三
    val rdd = sc.textFile("data/user_visit_action.txt")
    val actionRDD = rdd.map(
      line => {
        val datas = line.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    actionRDD.cache()

    val okIds = List(1L, 2L, 3L, 4L, 5L, 6L, 7L)
    val
    okIdsZip: List[(Long, Long)] = okIds.zip(okIds.tail)

    // TODO 0.计算分母（Word(页面)， count(数量)）
    val pageCntMap = actionRDD.filter(
      action => {
        okIds.init.contains(action.page_id)
      }
    ).map(
      action => {
        (action.page_id, 1)
      }
    ).reduceByKey(_ + _).collect.toMap

    // TODO 1. 将数据按照session进行分组
    val sessionGroupRDD: RDD[(String, Iterable[UserVisitAction])] =
      actionRDD.groupBy(_.session_id)

    // TODO 2. 分组后，对组内的数据按照时间进行排序（升序）
    val mapRDD: RDD[(String, List[(Long, Long)])] = sessionGroupRDD.mapValues(
      iter => {
        val sortActions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
        val sortIds: List[Long] = sortActions.map(_.page_id)
        // zip
        // 1,2,3,4,5,6,7
        // 2,3,4,5,6,7
        val zipIds: List[(Long, Long)] = sortIds.zip(sortIds.tail)

        //                val iterator: Iterator[List[Long]] = sortIds.sliding(2)
        //                iterator.map(
        //                    list => {
        //                        list(0)+"-"+list(1)
        //                    }
        //                )
        zipIds.filter(
          ids => {
            okIdsZip.contains(ids)
          }
        )
      }
    )
    // TODO 3.计算分子
    //      3.1 转换数据格式
    //      3.2 转换数据结构，将连续的数据转换为页面跳转路径
    //          3,5 => 3-5
    //      3.3 将连续的页面跳转路径统计数量
    // (sessionId,List(page_id,page_id))
    // list(page_id,page_id)
    val mapRDDs: RDD[List[(Long, Long)]] = mapRDD.map(_._2)
    // (page_id,page_id)
    val flatRDDs: RDD[(Long, Long)] = mapRDDs.flatMap(list => list)
    // ((page_id,page_id),1)
    // ((page_id,page_id)
    val reduceRDD: RDD[((Long, Long), Int)] = flatRDDs.map((_, 1)).reduceByKey(_ + _)
    val result: Array[((Long, Long), Int)] = reduceRDD.collect()

    // TODo 4. 计算单跳转换率
    //      分子 / 分母
    result.foreach {
      case ((pid1, pid2), sum) => {
        println(s"页面【${pid1} - ${pid2}】单跳转换率为 ：" + (sum.toDouble / pageCntMap.getOrElse(pid1, 0)))
      }
    }

    sc.stop()

  }

  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, //用户的ID
                              session_id: String, //Session的ID
                              page_id: Long, //某个页面的ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, //某一个商品品类的ID
                              click_product_id: Long, //某一个商品的ID
                              order_category_ids: String, //一次订单中所有品类的ID集合
                              order_product_ids: String, //一次订单中所有商品的ID集合
                              pay_category_ids: String, //一次支付中所有品类的ID集合
                              pay_product_ids: String, //一次支付中所有商品的ID集合
                              city_id: Long //城市 id
                            )
}
